#include "StdAfx.h"
#include "CommandManager.h"
#include "IOManager.h"
#include "Types.h"
#include "json.h"
#include "RandomGeneratedStreamInput.h"
#include "JsonStreamServer.h"
#include "SocketStreamInput.h"
#include "QueryIntermediateRepresentation.h"
#include "SocketStreamOutput.h"
#include "IStreamOutput.h"
#include "IStreamInput.h"
#include "ShowResultStreamOutput.h"
#include "QueryManager.h"
#include "PlanManager.h"
#include "Scheduler.h"
#include "WrapperManager.h"
CommandManager* CommandManager::commandManager = NULL;
CommandManager::CommandManager(void)
{
	initial();
}


CommandManager::~CommandManager(void)
{
}
CommandManager* CommandManager::getInstance()
{
	if(commandManager==NULL)
	{
		commandManager = new CommandManager();
	}
	return commandManager;
}
//use IOManager to listen to a port waiting for command
void CommandManager::initial()
{
	IOManager::getInstance()->addCommandInput(SERVER_IP,SERVER_PORT,this);
}
//the call back function called by IO manager, whenever new commands receive, this function would be called
//some kinds of command:
//1.register wrapper
//2.register query
bool CommandManager::processCommand(char* command,size_t length)
{
	std::string ret = "";
	int currentSize = 0;
		while(currentSize < length)
	{
		std::string cmd(command + currentSize);
		//std::cout<<cmd<<std::endl;
	    Document commandDocument = fromjson(cmd);
		
		std::string commandType = commandDocument.getField(COMMAND_TYPE).valuestr();
		if(commandType == REGISTER_WRAPPER)
		{
			processRegisterWrapperCommand(commandDocument);
		}
		else if(commandType == REGISTER_QUERY)
		{
			processRegisterQueryCommand(commandDocument);
		}
		else
		{
			assert(false); // never reached
			return false;
		}
	
		currentSize += cmd.length() + 1;
		
	}
	return true;
	 
}
bool CommandManager::processRegisterWrapperCommand(Document& document)
{
	Document& wrapperDocument = document.getField(WRAPPER_CONTENT).embeddedObject();
	WrapperManager::getInstance()->registerWrapper(wrapperDocument);
	return true;
}
bool CommandManager::processRegisterQueryCommand(Document& document)
{
	Document& queryDocument = document.getField(QUERY_CONTENT).embeddedObject();
	boost::shared_ptr<QueryIntermediateRepresentation>intermediateQuery(new QueryIntermediateRepresentation(queryDocument));
	Document& outputSpecificationDocument = document.getField(OUTPUT_SPECIFICATION).embeddedObject();
	std::string outputType =  outputSpecificationDocument.getField(OUTPUT_TYPE).valuestr();
	if(outputType == SHOW_RESULT)
	{
		boost::shared_ptr<IStreamOutput> streamOutput(new ShowResultStreamOutput());
		

		QueryManager::getInstance()->addStreamOutput(streamOutput); //register query
		PlanManager::getInstance()->generateQueryPlan(intermediateQuery); // generate plan
		Scheduler::getInstance()->setOperatorList(PlanManager::getInstance()->getOperatorList());// inform scheduler
	}
	else if(outputType == SOCKET_OUTPUT)
	{
		std::string ip = outputSpecificationDocument.getField(OUTPUT_IP).valuestr();
		std::string port = outputSpecificationDocument.getField(OUTPUT_PORT).valuestr();
		boost::shared_ptr<IStreamOutput> streamOutput(new SocketStreamOutput(ip,port));
		

		QueryManager::getInstance()->addStreamOutput(streamOutput);//register query
		PlanManager::getInstance()->generateQueryPlan(intermediateQuery);// generate plan
		Scheduler::getInstance()->setOperatorList(PlanManager::getInstance()->getOperatorList());// inform scheduler

	}
	else 
	{
		assert(false);//never reached
		return false;
	}
	return true;
}